package main

import (
	"fmt"
	"net"
)

type Server struct {
	broker Broker
	queue  Queue
}

type Message struct {
	id      uint32
	Topic   string `json:"topic"`
	Message string `json:"message"`
}

type BrokerCallback func(message Message)

type Broker interface {
	Pub(msg Message)
	Sub(topic string, callback BrokerCallback)
}

type Queue interface {
	Push(clientId int, msg Message)
	Pop(clientId int) Message
}

var clientId int

func (s *Server) handleConn(conn net.Conn) {
	myClientId := clientId + 1
	clientId += 1
	// 启动协程接收数据包
	go func(cid int, conn net.Conn) {
		for {
			_, msg, sign, err := ReadPacket(conn)
			if err != nil {
				return
			}
			switch sign {
			case SignClientPub:
				s.broker.Pub(msg)
			case SignClientSub:
				s.broker.Sub(msg.Topic, func(msg Message) {
					s.queue.Push(cid, msg)
				})
			}
		}
	}(myClientId, conn)
	// 启动协程消费消息
	go func(cid int) {
		for {
			msg := s.queue.Pop(cid)
			WritePacket(conn, msg.id, msg, SignComingMsg)
		}
	}(myClientId)
}

func (s *Server) Start(addr string) error {
	// 监听端口
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return err
	}
	fmt.Printf("SimpleMQ listen on %s\n", addr)
	go func() {
		// 对新客户端进行处理
		for {
			conn, err := ln.Accept()
			if err != nil {
				fmt.Println(err)
				continue
			}
			go s.handleConn(conn)
		}
	}()
	return nil
}
